package ds_industry_2025.ds.YangJuan_2024.tjxt


import org.apache.spark.ml.Pipeline
import org.apache.spark.ml.feature.{LabeledPoint, Normalizer, VectorAssembler}
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DoubleType
import org.apache.spark.ml.linalg.{DenseVector, Vector}

import java.util.Properties

/*
    根据子任务一的结果，计算出与用户id 为6708 的用户所购买相同商品种类最多的前 10
位用户 id（只考虑他俩购买过多少种相同的商品，不考虑相同的商品买了多少次），
并根据 Hive 的 dwd 库中相关表或 MySQL 数据库 shtd_store 中相关表，获取到这 10
位用户已购买过的商品，并剔除用户 6708 已购买的商品，通过计算这 10 位用户已购
买的商品（剔除用户 6708 已购买的商品）与用户 6708 已购买的商品数据集中商品的
余弦相似度累加再求均值，输出均值前 5 商品 id 作为推荐使用，将执行结果截图粘贴
至客户端桌面【Release\任务 C 提交结果.docx】中对应的任务序号下。
结果格式如下：
-----------------推荐 Top5 结果如下---------------
相似度 top1(商品 id：1，平均相似度：0.983456)
相似度 top2(商品 id：71，平均相似度：0.782672)
相似度 top3(商品 id：22，平均相似度：0.7635246)
相似度 top4(商品 id：351，平均相似度：0.7335748)
相似度 top5(商品 id：14，平均相似度：0.522356)
 */
object t1 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("T1")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    val conn=new Properties()
    conn.setProperty("user","root")
    conn.setProperty("password","123456")
    conn.setProperty("driver","com.mysql.jdbc.Driver")

    val order=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_info",conn)

    val detail=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_store?useSSL=false","order_detail",conn)

    //  todo 拿到初始数据
//    val source = order.join(detail, order("id") === detail("order_id"))
//      .select("user_id", "sku_id")
//      .distinct()

    val source = spark.table("tzgc.source").distinct()

    //  todo 先拿到用户6708购买的商品的数据
    val user_6708_skus:Array[Double] = source.filter(col("user_id") === 6708)
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()


    //  todo 拿到和6708购买相似的用户前10id
    val other_users = source.filter(col("user_id") !== 6708)
      .withColumn(
        "p",
        when(col("sku_id").cast(DoubleType).isin(user_6708_skus: _*), lit(1.0)).otherwise(lit(0.0))
      )
      .groupBy("user_id")
      .agg(sum("p").as("some"))
      .orderBy(desc("some"))
      .limit(10)
      .select("user_id")
      .map(_(0).toString.toDouble)
      .collect()

    //  todo 拿到前十用户购买的商品id
    val other_user_skus = source.filter(col("user_id").cast(DoubleType).isin(other_users: _*))
      .select("sku_id")
      .map(_(0).toString.toDouble)
      .collect()

    // todo 读取到特征工程第二题的把sku_info数据索引化之后的结果
    val sku_index = spark.table("tzgc.t1")

    //  todo 把数据的特征字段(除了id字段)转化为特征向量
    //  下面两种写法效果一样的
    val assembler = new VectorAssembler()
//      .setInputCols(sku_index.columns.slice(1,sku_index.columns.length))
      .setInputCols(sku_index.columns.toSeq.slice(1,sku_index.columns.toSeq.length).toArray)
      .setOutputCol("features")

    //  todo 设置机器学习流水线
    val pipeline = new Pipeline()
      .setStages(Array(assembler))
      .fit(sku_index).transform(sku_index)

    //  todo 把数据转化为labeledpoint方便后面进行分类
    val labelData:Dataset[LabeledPoint] = pipeline.select("id", "features").map(
      r => new LabeledPoint(r(0).toString.toInt, r(1).asInstanceOf[Vector])
    )

    //  todo 将数据进行标准化
    val normalizer = new Normalizer()
      .setInputCol("features")
      .setOutputCol("features_nor")
      .setP(2.0)

    val sku_info = normalizer.transform(labelData)
      .select("label","features_nor")
      .withColumnRenamed("features_nor","f")

    //  todo 定义计算余弦相似度的函数
    spark.udf.register(
      "cos",
      (v1:DenseVector,v2:DenseVector) => {
        1 - breeze.linalg.functions.cosineDistance(
          breeze.linalg.DenseVector(v1.values),
          breeze.linalg.DenseVector(v2.values)

        )
      }
    )

    // todo 聚合
    val result = sku_info.crossJoin(sku_info)
      .toDF("label", "f", "label2", "f2")
      .filter(col("label") !== col("label2"))
      .withColumn("cos", expr("cos(f,f2)"))
      .filter(col("label").isin(user_6708_skus: _*))
      .filter(!col("label2").isin(user_6708_skus: _*) && col("label2").isin(other_user_skus: _*))
      .groupBy("label2")
      .agg(avg("cos").as("avg_cos"))
      .orderBy(desc("avg_cos"))
      .limit(5)

    println("-----------推荐top5结果如下---------")
    result.collect().zipWithIndex.foreach{
      case(r,index) =>
        val id=r.getAs[Double]("label2").toInt
        val cos=r.getAs[Double]("avg_cos")
        val str=s"相似度top${index+1}(商品id:${id},平均相似度:${cos})"
        println(str)

    }










    spark.close()
  }

}
